package cmd

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"strconv"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/spf13/cobra"
	"github.com/tikv/client-go/txnkv"
)

var archiveCmd = &cobra.Command{
	Use:   "archive",
	Short: "Archive He3DB Xlog",
	Long:  "Welcome to use hbr for He3DB xlog archive",
	Run:   runArchive,
}

func init() {
	rootCmd.AddCommand(archiveCmd)
}

type Inode struct {
	Ino uint64 `json:"ino"`
}

func Unt64ToBytes(n uint64) []byte {
	x := uint64(n)
	bytesBuffer := bytes.NewBuffer([]byte{})
	binary.Write(bytesBuffer, binary.BigEndian, x)
	return bytesBuffer.Bytes()
}

func runArchive(cmd *cobra.Command, args []string) {
	access_key, _ := cmd.Flags().GetString("access_key")
	secret_key, _ := cmd.Flags().GetString("secret_key")
	endpoint, _ := cmd.Flags().GetString("endpoint")
	region, _ := cmd.Flags().GetString("region")
	bucket, _ := cmd.Flags().GetString("bucket")
	pd, _ := cmd.Flags().GetString("pd")
	backup_name, _ := cmd.Flags().GetString("name")
	archive_start_file, _ := cmd.Flags().GetString("archive_start_file")

	if access_key == "" || secret_key == "" || endpoint == "" || region == "" || bucket == "" || pd == "" || backup_name == "" || archive_start_file == "" {
		fmt.Printf("PARAMETER ERROR!\n")
		return
	}

	client, err := txnkv.NewClient([]string{pd})
	if err != nil {
		fmt.Printf("Connect Tikv Error!\n%v\n", err)
		return
	}
	txn, err := client.Begin()
	if err != nil {
		fmt.Printf("Tikv Transaction Begin Error!\n%v\n", err)
		return
	}

	sess, err := session.NewSession(&aws.Config{
		Region:      aws.String(region),
		Endpoint:    aws.String(endpoint),
		Credentials: credentials.NewStaticCredentials(access_key, secret_key, ""),
		S3ForcePathStyle: aws.Bool(true),
	})
	if err != nil {
		fmt.Printf("Connect S3 Error!\n%v\n", err)
		return
	}
	s3_client := s3.New(sess)

	var filename string = ""
	allCount := 0
	//1:meta
	fmt.Printf("Backup meta!\n")

	ret := make([]byte, 1)
	value, _ := strconv.ParseUint("00", 16, 8)
	ret[0] = byte(0xff & value)

	metaValue, err := txn.Get(context.TODO(), ret)
	if err != nil {
		fmt.Printf("Client Get meta Error!\n%v\n", err)
		return
	}

	filename = fmt.Sprintf("%x", ret)
	fmt.Printf("filename:%s\n", filename)

	_, err = s3_client.PutObject(&s3.PutObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(backup_name + "/" + filename),
		Body:   bytes.NewReader(metaValue),
	})
	if err != nil {
		fmt.Printf("S3 Put meta Error!\n%v\n", err)
		return
	}
	allCount++

	//2:pg_wal inode
	fmt.Printf("Backup pg_wal inode!\n")

	pwValue, err := txn.Get(context.TODO(), []byte("pg_wal"))
	if err != nil {
		fmt.Printf("Client Get pg_wal Error!\n%v\n", err)
		return
	}

	var pwi Inode
	json.Unmarshal(pwValue, &pwi)
	pwiKeyString := fmt.Sprintf("01%x", Unt64ToBytes(pwi.Ino))
	fmt.Printf("%v\n", pwiKeyString)
	ret = make([]byte, 9)
	index := 0
	for i := 0; i < len(pwiKeyString); i += 2 {
		value, _ := strconv.ParseUint(pwiKeyString[i:i+2], 16, 8)
		ret[index] = byte(0xff & value)
		index++
	}

	pwiValue, err := txn.Get(context.TODO(), ret)
	if err != nil {
		fmt.Printf("Client Get pwiValue Error!\n%v\n", err)
		return
	}

	filename = fmt.Sprintf("%x", ret)
	fmt.Printf("filename:%s\n", filename)

	_, err = s3_client.PutObject(&s3.PutObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(backup_name + "/" + filename),
		Body:   bytes.NewReader(pwiValue),
	})
	if err != nil {
		fmt.Printf("S3 Put pwiValue Error!\n%v\n", err)
		return
	}
	allCount++

	//3:pg_wal directory block
	fmt.Printf("Backup pg_wal directory block!\n")
	pwbKeyString := fmt.Sprintf("02%x0000000000000000", Unt64ToBytes(pwi.Ino))
	ret = make([]byte, 17)
	index = 0
	for i := 0; i < len(pwbKeyString); i += 2 {
		value, _ := strconv.ParseUint(pwbKeyString[i:i+2], 16, 8)
		ret[index] = byte(0xff & value)
		index++
	}

	pwbValue, err := txn.Get(context.TODO(), ret)
	if err != nil {
		fmt.Printf("Client Get pwbValue Error!\n%v\n", err)
		return
	}

	filename = fmt.Sprintf("%x", ret)
	fmt.Printf("filename:%s\n", filename)

	_, err = s3_client.PutObject(&s3.PutObjectInput{
		Bucket: aws.String(bucket), // bucket名称
		Key:    aws.String(backup_name + "/" + filename),
		Body:   bytes.NewReader(pwbValue),
	})
	if err != nil {
		fmt.Printf("S3 Put pwbValue Error!\n%v\n", err)
		return
	}
	allCount++
	//4:new file index
	fmt.Printf("Backup new file index!\n")
	pwiPrefixString := fmt.Sprintf("04%x", Unt64ToBytes(pwi.Ino))
	retStart := []byte(fmt.Sprintf("400000002%s", archive_start_file))
	retEnd := []byte("400000002ffffffffffffffffffffffff")
	index = 0
	for i := 0; i < len(pwiPrefixString); i += 2 {
		value, _ := strconv.ParseUint(pwiPrefixString[i:i+2], 16, 8)
		retStart[index] = byte(0xff & value)
		retEnd[index] = byte(0xff & value)
		index++
	}

	fiiter, err := txn.Iter(retStart, retEnd)
	if err != nil {
		fmt.Printf("new file index Iter Error!\n%v\n", err)
		return
	}

	newFileIndexCount := 0
	for fiiter.Valid() {
		k, v := fiiter.Key(), fiiter.Value()

		filename = fmt.Sprintf("%x", k)
		fmt.Printf("filename:%s\n", filename)

		_, err = s3_client.PutObject(&s3.PutObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(backup_name + "/" + filename),
			Body:   bytes.NewReader(v),
		})
		if err != nil {
			fmt.Printf("S3 PutObject Error!\n%v\n", err)
			return
		}

		if err := fiiter.Next(); err != nil {
			fmt.Printf("Iter Next Error!\n%v\n", err)
			return
		}
		newFileIndexCount++
		allCount++

		//5:new xlog inode
		fmt.Printf("Backup new xlog inode!\n")
		var wali Inode
		json.Unmarshal(v, &wali)
		waliKeyString := fmt.Sprintf("01%x", Unt64ToBytes(wali.Ino))
		ret = make([]byte, 9)
		index = 0
		for i := 0; i < len(waliKeyString); i += 2 {
			value, _ := strconv.ParseUint(waliKeyString[i:i+2], 16, 8)
			ret[index] = byte(0xff & value)
			index++
		}

		waliValue, err := txn.Get(context.TODO(), ret)
		if err != nil {
			fmt.Printf("Client Get waliValue Error!\n%v\n", err)
			return
		}

		filename = fmt.Sprintf("%x", ret)
		fmt.Printf("filename:%s\n", filename)

		_, err = s3_client.PutObject(&s3.PutObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(backup_name + "/" + filename),
			Body:   bytes.NewReader(waliValue),
		})
		if err != nil {
			fmt.Printf("S3 Put waliValue Error!\n%v\n", err)
			return
		}
		allCount++

		//6:new file block
		fmt.Printf("Backup new file block!\n")
		walbPrefixString := fmt.Sprintf("02%x", Unt64ToBytes(wali.Ino))
		retStartString := fmt.Sprintf("%s0000000000000000", walbPrefixString)
		retEndString := fmt.Sprintf("%s0000000000000100", walbPrefixString)

		retStart := make([]byte, 17)
		retEnd := make([]byte, 17)
		index = 0
		for i := 0; i < len(retStartString); i += 2 {
			value, _ := strconv.ParseUint(retStartString[i:i+2], 16, 8)
			retStart[index] = byte(0xff & value)
			value, _ = strconv.ParseUint(retEndString[i:i+2], 16, 8)
			retEnd[index] = byte(0xff & value)
			index++
		}

		walbIter, err := txn.Iter(retStart, retEnd)
		if err != nil {
			fmt.Printf("walbIter Error!\n%v\n", err)
			return
		}

		walbCount := 0
		for walbIter.Valid() {
			k1, v1 := walbIter.Key(), walbIter.Value()

			filename = fmt.Sprintf("%x", k1)
			fmt.Printf("filename:%s\n", filename)

			_, err = s3_client.PutObject(&s3.PutObjectInput{
				Bucket: aws.String(bucket), // bucket名称
				Key:    aws.String(backup_name + "/" + filename),
				Body:   bytes.NewReader(v1),
			})
			if err != nil {
				fmt.Printf("S3 PutObject Error!\n%v\n", err)
				return
			}

			if err := walbIter.Next(); err != nil {
				fmt.Printf("walbIter Next Error!\n%v\n", err)
				return
			}
			walbCount++
			allCount++
		}
		walbIter.Close()
		fmt.Printf("walbCount:%v\n", walbCount)
	}
	if err := txn.Commit(context.TODO()); err != nil {
		fmt.Printf("Tikv Transaction Commit Error!\n%v\n", err)
		return
	}
	fiiter.Close()
	fmt.Printf("new file index count:%v\n", newFileIndexCount)
	fmt.Printf("allCount:%v\n", allCount)
	client.Close()
}
